package com.bauer.mq.rabbitmq.config;

import java.util.HashMap;
import java.util.Map;
import javax.annotation.PostConstruct;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @description:
 * @author: weirui
 * @create: 2020-04-03 19:07
 **/
@Configuration
public class RabbitConfiguration {

    @Autowired
    private RabbitMQProperties rabbitMQProperties;

    @Bean
    public ConnectionFactory confirmConnection() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
        connectionFactory.setUsername(rabbitMQProperties.getUserName());
        connectionFactory.setPassword(rabbitMQProperties.getPassword());
        connectionFactory.setPort(rabbitMQProperties.getPort());
        connectionFactory.setHost(rabbitMQProperties.getHost());
        connectionFactory.setPublisherConfirmType(ConfirmType.SIMPLE);
        return connectionFactory;
    }

    @Bean(name = "rabbitListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
        @Qualifier("confirmConnection") ConnectionFactory firstConnectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(firstConnectionFactory);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setDefaultRequeueRejected(true);
        //开启监听者的数量  一个监听一个线程
        factory.setConcurrentConsumers(10);
        //每次获取的数量  配合concurrentConsumers 时需要适当的降低数量  要不就是一个线程跑多个数据
        factory.setPrefetchCount(1);
        return factory;
    }


    /**
     * 使用DirectRabbitListenerContainerFactory 时需要配置connectionFactory的并发线程池
     *
     * With the DirectMessageListenerContainer, you need to ensure that the connection factory is configured with a task
     * executor that had sufficient threads to support your desired concurrency across all listener containers that use
     * that factory. The default pool size is only five.
     */
    @Bean(name = "directRabbitListenerContainerFactory")
    public DirectRabbitListenerContainerFactory directRabbitListenerContainerFactory(
        @Qualifier("confirmConnection") ConnectionFactory firstConnectionFactory) {
        DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
        factory.setConnectionFactory(firstConnectionFactory);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setDefaultRequeueRejected(true);
        factory.setPrefetchCount(250);
        return factory;
    }


    @Bean
    public RabbitTemplate rabbitTemplate() {
        return new RabbitTemplate(confirmConnection());
    }

    @Bean
    public AmqpAdmin amqpAdmin(ConnectionFactory confirmConnection) {
        AmqpAdmin amqpAdmin = new RabbitAdmin(confirmConnection);
        return amqpAdmin;
    }
}
